package org.saxonextensions.eml.dispatcher;

import org.saxonextensions.eml.emitter.MimeMessageWrapper;

import javax.mail.*;
import javax.mail.internet.InternetAddress;
import java.io.InputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.RejectedExecutionException;
import java.util.logging.Level;
import java.util.logging.Logger;
import java.util.Arrays;

/**
 * Threaded dispatcher
 * <p/>
 * Date: 21-Dec-2005
 * Time: 10:47:41 AM
 *
 * @author IvanLatysh@gmail.com
 */
public abstract class AbstractThreadedEMLDispatcher implements EMLDispatcher {
  // Error code DTP00002

  // ----------------------------------------------------------------------------------
  /** counter */
  protected int count = 0;
  /**
   * Logger
   */
  protected transient Logger logger = Logger.getLogger(this.getClass().getName());
  /**
   * Thread pool
   */
  protected ExecutorService executor;
  /**
   * SMTP threads
   */
  protected int maxThreads = 100;

  /** Debug flag */
  protected boolean debug = false;

  /**
   * Return debug flag
   *
   * @return debug flag
   */
  public boolean getDebug() {
    synchronized (this) {
      return debug;
    }
  }

  public void setDebug(boolean debug) {
    synchronized (this) {
      this.debug = debug;
    }
  }

  public void increaseCounter() {
    synchronized (this) {
      count++;
    }
  }

  public void descreaseCounter() {
    synchronized (this) {
      count--;
    }
  }

  public int getCounter() {
    synchronized (this) {
      return count;
    }
  }

  public void setMaxThreads(int threads) {
    synchronized (this) {
      maxThreads = threads;
    }
  }

  public int getMaxThreads() {
    return maxThreads;
  }

  /**
   * Destroy thread pool
   */
  public void destroy() {
    if (null != executor) executor.shutdown();
  }

  /**
   * Return thread pool executor.
   *
   * @return executor or null when thread pool disabled
   */
  protected ExecutorService getExecutor() {
    if ((null == executor || executor.isShutdown()) && maxThreads > 0) {
      executor = Executors.newCachedThreadPool();
    }
    return executor;
  }

  /**
   * Process an input stream as serialized message
   *
   * @param in input stream with serialized message
   * @throws Exception when unable to process a message
   */
  public void dispatch(InputStream in) throws Exception {
    dispatch(new MimeMessageWrapper(getSession(), in));
  }

  /**
   * Return mail session
   *
   * @return session
   */
  public abstract Session getSession() throws Exception;

  /**
   * Send given message
   *
   * @param message message to send
   * @throws Exception when unable ro send the message
   */
  public void dispatch(final MimeMessageWrapper message) throws Exception {

    if (logger.isLoggable(Level.FINEST)) logger.log(Level.FINEST, "Pool size : " + getCounter());

    // sleep when pool is to large
    while (maxThreads > 0 && getCounter() >= maxThreads) {
      logger.log(Level.FINEST, "Thread pool size " + getCounter() + " >= " + maxThreads + ". Going to sleep.");
      synchronized (this) {
        try {
          wait(500);
        } catch (InterruptedException e) {
          logger.log(Level.WARNING, "Has been interrupted while sleeping.", e);
        }
      }
    }

    // construct the runnner
    final DispatcherRunner dispatcherRunner = new DispatcherRunner(message);

    try {
      // add sender to the pool
      if (maxThreads > 0) {
        logger.log(Level.FINEST, "Queuing message {" + message + "}. Pool size " + getCounter());
        // execute runner
        getExecutor().execute(dispatcherRunner);
        // increase runners count
        increaseCounter();
      } else {
        logger.log(Level.FINEST, "Pooling is disabled, sending message {" + message + "}");
        // simply send the message
        dispatcherRunner.run();
      }
    } catch (RejectedExecutionException e) {
      logger.log(Level.SEVERE, "Runner has been rejected.", e);
      throw e;
    }
  }

  /**
   * Dump given message
   * TODO move to a separate tools class
   *
   * @param message message to dump to file
   */
  public static void saveMessage(Message message) {
    final String filename = System.currentTimeMillis() + ".eml";
    if (null != message) {
      try {
        final File file = new File("eml-dump", filename);
        if (!file.getParentFile().exists()) file.getParentFile().mkdirs();
        file.createNewFile();
        message.writeTo(new FileOutputStream(file));
      } catch (Exception e) {
        e.printStackTrace();
      }
    }
  }

  /**
   * Dispatcher runner
   */
  class DispatcherRunner implements Runnable {
    MimeMessageWrapper msg;

    /**
     * Create a new runner with given message
     *
     * @param msg message
     */
    DispatcherRunner(MimeMessageWrapper msg) {
      this.msg = msg;
    }

    public void run() {
      try {
        long sendTime;
        final long sendStart = System.currentTimeMillis();
        Transport tr = null;
        Session session = null;

        try {
          session = msg.getSession();
          // set debug
          session.setDebug(getDebug());
          // get transport
          tr = session.getTransport("smtp");

          int tryCount = 0;

          try {
            tr.connect();
          } catch (MessagingException e) {
            logger.log(Level.SEVERE, "Unable to connect to SMTP server. Will try send the message {" + msg + "} anyway.", e);
          }

          tryCount = 0;
          boolean sent = false;
          while (!sent) {
            try {
              // send message
              tr.sendMessage(msg.getMessage(), msg.getAllRecipients());
              // set send flag
              sent = true;
            } catch (IllegalStateException e) {
              // the connection can be lost. Try to reconnect.
              logger.log(Level.SEVERE, "Error while sending the message {" + msg + "}, will try to reconnect.", e);
              if (tryCount < 3 && !tr.isConnected()) {
                synchronized (this) {
                  try {
                    wait(10000);
                  } catch (InterruptedException e1) {
                    logger.log(Level.SEVERE, "Interrupted", e1);
                  }
                }
                tr.connect();
                tryCount++;
              } else {
                logger.log(Level.SEVERE, "Unable to send the message {" + msg + "}, tried {" + tryCount + "} times.", e);
                throw e;
              }
            }
          }
          sendTime = System.currentTimeMillis() - sendStart;
          // log message sent email
          if (logger.isLoggable(Level.FINEST)) {
            logger.info("Message to " + Arrays.toString(msg.getAllRecipients()) + " send successfully. In " + sendTime + "ms.");
          }
        } catch (Exception e) {
          String rcp = "";
          try {
            if (null != msg && null != msg.getAllRecipients()) {
              for (Address recipient : msg.getAllRecipients()) {
                rcp += "[" + ((InternetAddress) recipient).getAddress() + "]";
              }
            }
          } catch (MessagingException e1) {
            logger.log(Level.SEVERE, "Unable retrieve recipients from the message for logging.", e1);
          }
          logger.log(Level.SEVERE, "Unable to sent the message to {" + rcp + "}, saving eml into a temp file.", e);
          saveMessage(msg.getMessage());
        } finally {
          try {
            tr.close();
          } catch (MessagingException e) {
            logger.log(Level.SEVERE, "Unable to close Transport", e);
          }
        }
      } catch (Exception ex) {
        logger.log(Level.SEVERE, "[DTP00001] Error during eMail dispatch.", ex);
      } finally {
        descreaseCounter();
      }
    }
  }

}
